package com.redisdemo.demo.stream;

import io.lettuce.core.*;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisStreamCommands;

import java.util.List;

public class Test1 {
    public static void main(String[] args) {
        RedisURI redisUri = RedisURI.create("localhost", 6379);
        RedisClient redisClient = RedisClient.create(redisUri);
        StatefulRedisConnection<String, String> connection = redisClient.connect();
        RedisStreamCommands<String, String> streamCommands = connection.sync();

//        streamCommands.xread(XReadArgs.StreamOffset.latest("first-stream"));

        List<StreamMessage<String, String>> messages = streamCommands.xreadgroup(
                Consumer.from("lettuce-group", "线程1"),
                XReadArgs.Builder.block(1000).count(1000),
                XReadArgs.StreamOffset.from("first-stream", "0-0"),
                XReadArgs.StreamOffset.lastConsumed("first-stream"));
        if (messages != null & messages.size() > 0) {
            StringBuffer sb=new StringBuffer();
            for (StreamMessage<String, String> message : messages) {
                System.out.println("线程1" +message.getId() + "：" + message.getBody());
                streamCommands.xack("first-stream", "lettuce-group",  message.getId());
            }
        }else {
            System.out.println("没有消费");
        }

        connection.close();
        redisClient.shutdown();
//
//        new Thread(new Runnable() {
//            @Override
//            public void run() {
//                while (true){
//
//                }
//            }
//        }).run();
    }
}
